use crate::vdd_prometheus::MODEL_COUNT_GAUGE;
use crate::{extract_field_ids, vdd_polars, JwtClaims, WebResponseSupportTrait};
use chrono::Local;
use polars::prelude::{ParquetReader, SerReader};
use salvo::fs::NamedFile;
use salvo::jwt_auth::JwtAuthDepotExt;
use salvo::{handler, Depot, Request, Response};
use serde::Serialize;
use serde_json::Value;
use std::collections::HashMap;
use std::fs::File;
use std::str::FromStr;
use std::sync::{Arc, LazyLock};
use tokio::fs;
use visit_dd_core::i18n_text::{E_ARG_BAD, E_NO_PERMISSION_FOR_ACTION};
use visit_dd_core::model::{DataStore, VddModel, VddModelRef};
use visit_dd_core::proj::APP_CONFIG;
use visit_dd_core::{i18n, Astr, FieldId, JsonSupportTrait};
use visit_dd_db::sqlx::SqlxDataStoreTrait;
use visit_dd_db::{cache, field_db, model_db, record_rule_db, visit_db};
use visit_dd_db::{record_db, subscribe};

/// 解析刚刚上传的parquet文件
#[handler]
pub(crate) async fn p_handle_parquet(req: &mut Request, _depot: &mut Depot, res: &mut Response) {
    let file_key = req.param::<&str>("file-key").unwrap();
    let path = format!("{}/{}", APP_CONFIG.upload_dir, file_key);
    let file = File::open(&path).unwrap();
    let reader = ParquetReader::new(file);
    let df = reader.finish();
    if df.is_err() {
        res.error(&df.unwrap_err().to_string());
        return;
    }
    let df = df.unwrap();
    vdd_polars::df_cache_put(u64::from_str(file_key).unwrap(), &df);
    res.ok(&df.shape());
}

/// 客户端删除已上传但又不想要的parquet文件
#[handler]
pub(crate) async fn p_drop_parquet(req: &mut Request, _depot: &mut Depot, res: &mut Response) {
    let file_key = req.param::<&str>("file-key").unwrap();
    let path = format!("{}/{}", APP_CONFIG.upload_dir, file_key);
    fs::remove_file(path).await.unwrap();
    vdd_polars::df_cache_remove(u64::from_str(file_key).unwrap());
    res.ok_empty();
}

#[handler]
pub(crate) async fn p_create_model(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j = req.parse_body::<Value>().await.unwrap();
    let model_count = model_db::m_count_all().await;
    let allow_visit = j["allowVisit"].as_bool().unwrap_or(false);
    let model = if j["store"].is_object() {
        let ds = DataStore::new(&j["store"], &APP_CONFIG.pri_key, &APP_CONFIG.pub_key);
        let Ok(ds) = ds else {
            res.error(&ds.unwrap_err());
            return;
        };
        if !ds.test_connection().await {
            res.error(i18n!(1201).as_ref());
            return;
        }
        let mut m = VddModel::new(&ds.default_model_name(), &jwt.uid, "", allow_visit);
        let fields = ds.import_fields(&m).await;
        m.store = Some(Arc::new(ds));
        model_db::m_save(&m, false).await.unwrap();
        field_db::f_clear_in_model(m.id, &jwt.uid).await.unwrap(); // 确保模型内字段清空
        field_db::f_save_some(&fields, &jwt.uid).await.unwrap();
        let mut tag_map = HashMap::<Astr, u64>::new();
        fields.iter().for_each(|field| {
            tag_map.insert(field.label.replace(":", "").into(), field.id.0);
        });
        model_db::m_save_tags(&m, &m.name, &tag_map).await.unwrap();
        m
    } else {
        let m = VddModel::new(
            j["name"].as_str().expect(i18n!(E_ARG_BAD).as_ref()),
            &jwt.uid,
            j["memo"].as_str().unwrap_or(""),
            allow_visit,
        );
        model_db::m_save(&m, false).await.unwrap();
        m
    };
    LazyLock::force(&MODEL_COUNT_GAUGE).set((model_count + 1) as i64);
    res.ok(&model);
}

#[handler]
pub(crate) async fn p_save_model(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let model_id = j["id"].as_u64().unwrap();
    let mut model = model_db::m_load(model_id).await;
    let model = Arc::get_mut(&mut model).unwrap();
    if model.uid != jwt.uid {
        res.error(i18n!(E_NO_PERMISSION_FOR_ACTION).as_ref());
        return;
    }
    let mut changed = false;
    if let Some(x) = j["name"].as_str() {
        model.name = x.into();
        changed = true;
    };
    if let Some(x) = j["memo"].as_str() {
        model.memo = Some(x.into());
        changed = true;
    }
    if let Some(x) = j["allowVisit"].as_bool() {
        model.allow_visit = x;
        changed = true;
    }
    // 检测数据源配置信息是否有变动
    if model.store.is_some() && j["store"].as_object().is_some() {
        let ds_new = DataStore::new(&j["store"], &APP_CONFIG.pri_key, &APP_CONFIG.pub_key);
        match ds_new {
            Ok(ds_new) => {
                if !ds_new.test_connection().await {
                    res.error(i18n!(1201).as_ref());
                    return;
                }
                if !model.store.clone().unwrap().same_as(&ds_new) {
                    model.store = Some(Arc::new(ds_new));
                    changed = true;
                }
            }
            Err(e) => {
                res.error(&e);
                return;
            }
        }
    }
    if changed {
        model.update_ts = Local::now().naive_local();
        res.wrap(model_db::m_save(&model, true).await);
    } else {
        res.ok_empty();
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct ListMyModelResponse {
    pub model: Arc<VddModel>,
    pub field_count: u64,
    pub req_sum: u16,
    pub req_handled: u16,
    pub rule_count: usize,
    pub vg_count: usize,
    pub subscription_count: usize,
}

#[handler]
pub(crate) async fn g_mine(_req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let mut model_info_vec = Vec::<ListMyModelResponse>::new();
    let model_vec = model_db::m_select_by_uid(&jwt.uid).await;
    let model_ids = model_vec.iter().map(|m| m.id).collect::<Vec<u64>>();
    let rule_counts = record_rule_db::count_rule(&model_ids).await;
    let vg_counts = visit_db::vg_count_in_models(&model_ids).await;
    // println!("vg_counts: {:?}", vg_counts);
    let subscription_counts = subscribe::sc_count_in_models(&model_ids).await;
    // tracing::info!("model_vec: {:?}", model_vec);
    let mut i = 0;
    for model in model_vec {
        let c = visit_db::vr_count_in_model(model.id).await.unwrap();
        let s = cache::get_schema(model.id).await.unwrap();
        model_info_vec.push(ListMyModelResponse {
            model: model.clone(),
            field_count: s.fields.len() as u64,
            req_sum: c.0 as u16,
            req_handled: c.1 as u16,
            rule_count: rule_counts[i],
            vg_count: vg_counts[i],
            subscription_count: subscription_counts[i],
        });
        i += 1;
    }
    res.ok(&model_info_vec);
}

#[handler]
pub(crate) async fn p_remove(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j = req.parse_body::<Value>().await.unwrap();
    if !j.has_useful_int("modelId") {
        res.error(&i18n!(E_ARG_BAD));
        return;
    }
    let model_id = j["modelId"].as_u64().unwrap();
    let s = cache::get_schema(model_id).await.unwrap();
    if s.fields.len() > 0 {
        res.error(&i18n!(1018));
    } else {
        if s.model.uid != jwt.uid {
            res.error(i18n!(E_NO_PERMISSION_FOR_ACTION).as_ref());
        } else {
            let r = model_db::m_delete(&s.model, &jwt.uid).await;
            res.wrap(r);
        }
    }
}

#[handler]
pub(crate) async fn p_export(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j = req.parse_body::<Value>().await.unwrap();
    let model_id = j["modelId"].as_u64();
    match model_id {
        None => {
            res.error(i18n!(E_ARG_BAD).as_ref());
            return;
        }
        Some(model_id) => {
            let schema = cache::get_schema(model_id).await.unwrap();
            if schema.model.uid != jwt.uid {
                res.error("不是自己的模型，无法操作");
                return;
            }
            let model_code = schema.tags.get(&model_id);
            match model_code {
                None => {
                    res.error(i18n!(1019).as_ref());
                }
                Some(model_code) => {
                    // 开始导出数据
                    let r = if j["withData"].as_bool().unwrap_or(false) {
                        let records = record_db::select_records_all(model_id).await;
                        vdd_polars::export_with_record(&schema, &records).await
                    } else {
                        vdd_polars::export_schema(&schema)
                    };
                    match r {
                        Ok(_) => {
                            res.ok(&model_code);
                        }
                        Err(e) => {
                            res.error(&e);
                        }
                    }
                }
            }
        }
    }
}

/// 单纯的下载文件 因为axios发送的请求分为是否是blob请求，如果是，会将后台的消息返回为blob，不解析，影响错误消息展示
#[handler]
pub(crate) async fn p_download_parquet(req: &mut Request, _depot: &mut Depot, res: &mut Response) {
    let j = req.parse_body::<Value>().await.unwrap();
    let model_id = j["modelId"].as_u64().unwrap();
    let path = vdd_polars::parquet_cache_path(model_id);
    NamedFile::builder(path).send(req.headers(), res).await;
}

/// 上传parquet文件
#[handler]
pub(crate) async fn p_import(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j = req.parse_body::<Value>().await.unwrap();
    let model_id = j["modelId"].as_u64();
    if model_id.is_none() {
        res.error(i18n!(E_ARG_BAD).as_ref());
        return;
    }
    let model_id = model_id.unwrap();
    let schema = cache::get_schema(model_id).await.unwrap();
    let mut fields_latest = schema.fields.clone();
    let files = j["files"].as_array();
    if files.is_none() {
        res.error(i18n!(E_ARG_BAD).as_ref());
        return;
    }
    let file = files.unwrap().first().unwrap().as_u64().unwrap_or(0);
    if file < 1 {
        res.error(i18n!(E_ARG_BAD).as_ref());
        return;
    }
    let df = vdd_polars::df_cache_get(file).unwrap();
    let force_fields = j["forceFields"].as_bool().unwrap_or(false);
    if force_fields {
        field_db::f_clear_in_model(model_id, &jwt.uid)
            .await
            .unwrap(); // 确保模型内字段清空
        let fields = vdd_polars::fields_from_df(model_id, &df.df).unwrap();
        let r = field_db::f_save_some(&fields, &jwt.uid).await;
        if r.is_err() {
            res.error(&r.unwrap_err());
            return;
        };
        fields_latest = fields;
    }
    let records = vdd_polars::records_from_df(model_id, jwt.uid.clone(), &fields_latest, &df.df);
    match records {
        Err(e) => {
            res.error(&e);
        }
        Ok(records) => {
            let clean_records = j["matchRecord"].as_str().unwrap_or("keep") == "clear";
            if clean_records {
                record_db::clear_records(model_id, jwt.uid.clone())
                    .await
                    .unwrap();
            }
            record_db::append_from_parquet(&schema, &records).unwrap();
            res.ok_empty();
        }
    }
}

#[handler]
pub(crate) async fn p_drop(req: &mut Request, _depot: &mut Depot, res: &mut Response) {
    let j = req.parse_body::<Value>().await.unwrap();
    if !j.has_useful_int("modelId") {
        res.error(&i18n!(E_ARG_BAD));
        return;
    }
    let model_id = j["modelId"].as_u64().unwrap();
    let ares = model_db::m_drop_from_trash(model_id).await;
    res.wrap(ares);
}

#[handler]
pub(crate) async fn p_restore(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req
        .parse_body::<Value>()
        .await
        .expect(i18n!(E_ARG_BAD).as_ref());
    let model_id = j["modelId"].as_u64().unwrap();
    let field_ids = extract_field_ids(&j);
    let with_values = j["withValues"].as_bool().unwrap_or(false);
    let ares = model_db::m_restore(model_id, &field_ids, with_values, jwt.uid.clone()).await;
    res.wrap(ares);
}

/// 获取链接到某个字段的所有(在一个用户名下的)模型列表
#[handler]
pub(crate) async fn g_models_use_field(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let f_id = req.params().get("fid").cloned().unwrap();
    let f_id = u64::from_str(&f_id).unwrap();
    let f_id = FieldId(f_id);
    let models = model_db::m_select_by_uid(&jwt.uid).await;
    let mut model_vec = Vec::<VddModelRef>::new();
    let mut found_field = false;
    for model in models {
        let fields = field_db::f_select_by_model(model.id).await;
        for field in fields {
            if !found_field {
                found_field = field.id == f_id;
            }
            if field.used_field(&f_id) {
                model_vec.push(model.clone());
            }
        }
    }
    if !found_field {
        res.error("此字段不在作者名下");
        return;
    } else {
        res.ok(&model_vec);
    }
}
